package com.ferry.transport.netty.handler.client;

import com.ferry.protocol.payload.ResponsePayload;
import com.ferry.transport.api.channel.WrappedChannel;
import com.ferry.transport.netty.concurrent.CloseableExecutor;
import com.ferry.transport.netty.concurrent.ExecutorFactory;
import com.ferry.transport.netty.concurrent.ThreadPoolExecutorFactory;
import com.ferry.transport.netty.consumer.Response;

/**
 * Created by daiyong
 */
public class DefaultConsumerProcessor implements ConsumerProcessor{

	private CloseableExecutor executor;

	public DefaultConsumerProcessor() {
		this(new ThreadPoolExecutorFactory().newExecutor(ExecutorFactory.Target.CONSUMER, "ferry-consumer"));
	}

	public DefaultConsumerProcessor(CloseableExecutor executor) {
		this.executor = executor;
	}

	@Override
	public void handleResponse(WrappedChannel channel, ResponsePayload responsePayload) throws Exception {

		Response response = new Response();
		response.setResponsePayload(responsePayload);
		ConsumerMessageTask task = new ConsumerMessageTask(channel, response);
		executor.execute(task);
	}

	@Override
	public void shutdown() {
		executor.shutdown();
	}
}
